package utils;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class HdfsUtils {
	public static Path getPath(FileSystem fs, String path, String name) {
		try {
			FileStatus[] result = fs.listStatus(new Path(path));
			for (int i = 0; i < result.length; i++) {
				FileStatus item = result[i];
				if (item.isFile()) {
					if (item.getPath().getName().endsWith(name))
						return item.getPath();
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

	public static void readPath(FileSystem fs, String path, String name, IReadFile readFile) {
		try {
			FileStatus[] result = fs.listStatus(new Path(path));
			for (int i = 0; i < result.length; i++) {
				FileStatus item = result[i];
				if (item.isFile()) {
					if (item.getPath().getName().endsWith(name)) {
						try {
							BufferedReader fis = new BufferedReader(new InputStreamReader(fs.open(item.getPath())));
							String line = null;
							while ((line = fis.readLine()) != null) {
								line = line.trim();
								readFile.paserLine(CSVFileUtil.parseLine(line));
							}
							fis.close();
						} catch (Exception e) {
						}
						break;
					}
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static Map<String, String> readPath(FileSystem fs, String path, String name) {
		Map<String, String> map = new HashMap<String, String>();
		try {
			FileStatus[] result = fs.listStatus(new Path(path));
			for (int i = 0; i < result.length; i++) {
				FileStatus item = result[i];
				if (item.isFile()) {
					if (item.getPath().getName().endsWith(name)) {
						try {
							BufferedReader fis = new BufferedReader(new InputStreamReader(fs.open(item.getPath())));
							String line = null;
							while ((line = fis.readLine()) != null) {
								line = line.trim();
								String[] datas = line.split("=");
								if (datas.length == 2) {
									map.put(datas[0], datas[1]);
								}
							}
							fis.close();
						} catch (Exception e) {
						}
						break;
					}
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		return map;
	}
}
